/**
 * ## 编程作业
      实时轨迹场景中使用Redis存储了当前车辆的实时经纬度数据，请使用Mysql替换Redis来存储实时经纬度数据。
      需要开发代码，使用Spark Structured Streaming把实时经纬度数据写入Mysql.
 */

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object StructuredKafka {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)
    //1.获取sparksession
    val spark: SparkSession = SparkSession.builder()
      .master("local[*]")
      .appName(StructuredKafka.getClass.getName)
      .getOrCreate()
    import spark.implicits._

    //2.读取kafka数据源
    val kafkaDF: DataFrame = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "linux121:9092")
      .option("subscribe", "lagou_bus_info")
      .load()

    //3.处理数据
    val df: DataFrame = kafkaDF.selectExpr("CAST(value AS STRING)")

    //4.df 转为 ds
    val ds: Dataset[String] = df.as[String]

    //5.数据封装为BusInfo
    val res: Dataset[BusInfo] = ds.map(BusInfo(_)).filter(_ != null)

    //6.数据写入MySQL
    res.writeStream
      .foreach(new MySQLWriter)
      .outputMode("append")
      .start()
      .awaitTermination()
  }
}
